//===----------------------------------------------------------------------===//
//
//                         BusTub
//
// buffer_pool_manager.cpp
//
// Identification: src/buffer/buffer_pool_manager.cpp
//
// Copyright (c) 2015-2021, Carnegie Mellon University Database Group
//
//===----------------------------------------------------------------------===//

#include "buffer/buffer_pool_manager.h"

#include "common/exception.h"
#include "common/macros.h"
#include "storage/page/page_guard.h"

namespace bustub {

BufferPoolManager::BufferPoolManager(size_t pool_size, DiskManager *disk_manager, size_t replacer_k,
                                     LogManager *log_manager)
    : pool_size_(pool_size), disk_manager_(disk_manager), log_manager_(log_manager) {
  // we allocate a consecutive memory space for the buffer pool
  pages_ = new Page[pool_size_];
  replacer_ = std::make_unique<LRUKReplacer>(pool_size, replacer_k);

  // Initially, every page is in the free list.
  for (size_t i = 0; i < pool_size_; ++i) {
    free_list_.emplace_back(static_cast<int>(i));
  }
}

BufferPoolManager::~BufferPoolManager() { delete[] pages_; }

auto BufferPoolManager::NewPage(page_id_t *page_id) -> Page * {
  latch_.lock();
  frame_id_t allocated_id;
  if (!free_list_.empty()) {
    // Ok, wo have free page to allocate
    *page_id = AllocatePage();
    allocated_id = free_list_.front();
    free_list_.pop_front();
  } else {
    if (replacer_->Evict(&allocated_id)) {
      if (pages_[allocated_id].IsDirty()) {
        disk_manager_->WritePage(pages_[allocated_id].page_id_, pages_[allocated_id].GetData());
        pages_[allocated_id].is_dirty_ = false;
      }
      page_table_.erase(pages_[allocated_id].page_id_);
      *page_id = AllocatePage();
    } else {
      latch_.unlock();
      return nullptr;
    }
  }
  ReInitializePage(*page_id, allocated_id);
  latch_.unlock();
  return &pages_[allocated_id];
}

auto BufferPoolManager::FetchPage(page_id_t page_id, [[maybe_unused]] AccessType access_type) -> Page * {
  latch_.lock();
  if (page_table_.find(page_id) != page_table_.end()) {
    frame_id_t frame_id = page_table_[page_id];
    pages_[frame_id].pin_count_++;
    replacer_->SetEvictable(frame_id, false);
    replacer_->RecordAccess(frame_id);
    latch_.unlock();
    return &pages_[frame_id];
  }
  frame_id_t allocated_id;
  if (!free_list_.empty()) {
    // Ok, wo have free page to allocate
    allocated_id = free_list_.front();
    free_list_.pop_front();
  } else {
    if (replacer_->Evict(&allocated_id)) {
      if (pages_[allocated_id].IsDirty()) {
        disk_manager_->WritePage(pages_[allocated_id].page_id_, pages_[allocated_id].GetData());
        pages_[allocated_id].is_dirty_ = false;
      }
      page_table_.erase(pages_[allocated_id].page_id_);
    } else {
      latch_.unlock();
      return nullptr;
    }
  }
  ReInitializePage(page_id, allocated_id);
  disk_manager_->ReadPage(page_id, pages_[allocated_id].GetData());
  latch_.unlock();
  return &pages_[allocated_id];
}

auto BufferPoolManager::UnpinPage(page_id_t page_id, bool is_dirty, [[maybe_unused]] AccessType access_type) -> bool {
  latch_.lock();
  if (page_table_.find(page_id) == page_table_.end()) {
    // page not exists
    latch_.unlock();
    return false;
  }
  frame_id_t frame_id = page_table_[page_id];
  if (pages_[frame_id].pin_count_ == 0) {
    latch_.unlock();
    return false;
  }
  pages_[frame_id].pin_count_--;
  if (pages_[frame_id].pin_count_ == 0) {
    replacer_->SetEvictable(frame_id, true);
  }
  pages_[frame_id].is_dirty_ |= is_dirty;
  latch_.unlock();
  return true;
}

auto BufferPoolManager::FlushPage(page_id_t page_id) -> bool {
  latch_.lock();
  if (page_id == INVALID_PAGE_ID || page_table_.find(page_id) == page_table_.end()) {
    latch_.unlock();
    return false;
  }
  frame_id_t frame_id = page_table_[page_id];
  disk_manager_->WritePage(page_id, pages_[frame_id].GetData());
  pages_[frame_id].is_dirty_ = false;
  latch_.unlock();
  return true;
}

void BufferPoolManager::FlushAllPages() {
  latch_.lock();
  auto it = page_table_.begin();
  auto end = page_table_.end();
  for (; it != end; it++) {
    if (pages_[it->second].IsDirty()) {
      disk_manager_->WritePage(it->first, pages_[it->second].GetData());
      pages_[it->second].is_dirty_ = false;
    }
  }
  latch_.unlock();
}

auto BufferPoolManager::DeletePage(page_id_t page_id) -> bool {
  latch_.lock();
  if (page_id == INVALID_PAGE_ID || page_table_.find(page_id) == page_table_.end()) {
    latch_.unlock();
    return true;
  }
  frame_id_t frame_id = page_table_[page_id];
  if (pages_[frame_id].GetPinCount() > 0) {
    latch_.unlock();
    return false;
  }
  if (pages_[frame_id].IsDirty()) {
    disk_manager_->WritePage(page_id, pages_[frame_id].GetData());
    pages_[frame_id].is_dirty_ = false;
  }
  page_table_.erase(page_id);
  replacer_->Remove(frame_id);
  free_list_.emplace_back(static_cast<int>(frame_id));
  pages_[frame_id].ResetMemory();
  pages_[frame_id].pin_count_ = 0;
  pages_[frame_id].is_dirty_ = false;
  pages_[frame_id].page_id_ = INVALID_PAGE_ID;
  DeallocatePage(page_id);
  latch_.unlock();
  return true;
}

auto BufferPoolManager::AllocatePage() -> page_id_t { return next_page_id_++; }

auto BufferPoolManager::FetchPageBasic(page_id_t page_id) -> BasicPageGuard { return {this, FetchPage(page_id)}; }

auto BufferPoolManager::FetchPageRead(page_id_t page_id) -> ReadPageGuard {
  Page *pg = FetchPage(page_id);
  if (pg != nullptr) {
    pg->RLatch();
    return {this, pg};
  }
  return {this, nullptr};
}

auto BufferPoolManager::FetchPageWrite(page_id_t page_id) -> WritePageGuard {
  Page *pg = FetchPage(page_id);
  if (pg != nullptr) {
    pg->WLatch();
    return {this, pg};
  }
  return {this, nullptr};
}

auto BufferPoolManager::NewPageGuarded(page_id_t *page_id) -> BasicPageGuard { return {this, NewPage(page_id)}; }

void BufferPoolManager::ReInitializePage(page_id_t page_id, frame_id_t allocated_id) {
  page_table_[page_id] = allocated_id;
  pages_[allocated_id].ResetMemory();
  pages_[allocated_id].page_id_ = page_id;
  pages_[allocated_id].is_dirty_ = false;
  pages_[allocated_id].pin_count_ = 1;
  replacer_->RecordAccess(allocated_id);
  replacer_->SetEvictable(allocated_id, false);
}
}  // namespace bustub
